package org.mq.mymq.nameserver.event.listener;

import com.alibaba.fastjson.JSON;
import io.netty.util.internal.StringUtil;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.enums.NameServerEventCode;
import org.mq.mymq.common.event.Listener;
import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.common.TraceReplicationProperties;
import org.mq.mymq.nameserver.event.model.NodeReplicationAckMsgEvent;
import org.mq.mymq.nameserver.event.model.NodeReplicationMsgEvent;
import org.mq.mymq.nameserver.event.model.ReplicationMsgEvent;
import org.mq.mymq.nameserver.store.ServiceInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.Inet4Address;

/**
 * @ClassName NodeReplicationMsgLIstener
 * @Description 接收上一个节点同步的复制数据内容
 * @Author jiarong_ye
 * @Date 2025/6/25 10:36
 * @Version 1.0
 */
public class NodeReplicationMsgListener implements Listener<NodeReplicationMsgEvent> {
    private static final Logger log = LoggerFactory.getLogger(NodeReplicationMsgListener.class);

    @Override
    public void onReceive(NodeReplicationMsgEvent event) throws Exception {
        ServiceInstance serviceInstance = event.getServiceInstance();
        // 接收到上一个节点同步的数据，存入本地内存
        CommonCache.getServiceInstanceManager().put(serviceInstance);
        // 将数据发送给下一个节点
        ReplicationMsgEvent replicationMsgEvent = new ReplicationMsgEvent();
        replicationMsgEvent.setServiceInstance(serviceInstance);
        replicationMsgEvent.setMsgId(event.getMsgId());
        replicationMsgEvent.setType(event.getType());
        CommonCache.getReplicationMsgQueueManager().put(replicationMsgEvent);
        // 如果下一个节点是尾部节点，不需要给下一个节点发送数据，但是需要返回一个ack给上一个节点
        TraceReplicationProperties traceReplicationProperties = CommonCache.getNameserverProperties().getTraceReplicationProperties();
        String nextNode = traceReplicationProperties.getNextNode();
        if (StringUtil.isNullOrEmpty(nextNode)) {
            NodeReplicationAckMsgEvent nodeReplicationAckMsgEvent = new NodeReplicationAckMsgEvent();
            nodeReplicationAckMsgEvent.setNodeIp(Inet4Address.getLocalHost().getHostAddress());
            nodeReplicationAckMsgEvent.setNodePort(traceReplicationProperties.getPort());
            nodeReplicationAckMsgEvent.setType(replicationMsgEvent.getType());
            nodeReplicationAckMsgEvent.setMsgId(replicationMsgEvent.getMsgId());
            CommonCache.getPreNodeChannel().writeAndFlush(new TcpMsg(NameServerEventCode.NODE_REPLICATION_ACK_MSG.getCode(),
                    JSON.toJSONBytes(nodeReplicationAckMsgEvent)));
            log.info("尾节点ack信号");
        }

    }
}
